<?php
namespace xlx;

//修复workerman-windows
if (strpos(strtolower(PHP_OS), 'win') === 0 && !class_exists(\Windows\Worker::class,false)) {
    class_alias(\Windows\Worker::class, \Workerman\Worker::class);
    class_alias(\Windows\Events\Select::class, \Workerman\Events\Select::class); 
}

class DispatchServer
{
    public $worker;
    public $option=[];
    public function __construct($opt = [ 'ip_addr'=>'127.0.0.1', 'ip_port'=>8109,])
    {
        $w = new \Workerman\Worker("tcp://{$opt['ip_addr']}:{$opt['ip_port']}");
        $w->name='DispatchServer';
        $w->protocol = \Protocols\Json::class;
        $w->onWorkerStart=[$this,'onWorkerStart'];
        $w->onConnect=[$this,'onConnect'];
        $w->onMessage=[$this,'onMessage'];
        $w->onClose=[$this,'onClose'];
        $this->worker = $w;
        $this->option = $opt + [ #default value
            'proc_max'=>$cpu=Util::getCoreCount()*2,
            'proc_min'=>0,//max($cpu/4,1),
            'proc_ttl'=>12,
            //'profile'=>'nil',
        ];

        $this->queue_tasks = new \SplDoublyLinkedList();
        $this->idle_workers = new \SplDoublyLinkedList();
    }

    public $time;
    public $worker_count = 0;
    public $idle_workers= null;//[ conn,conn,...] //可用进程
    public $queue_tasks = null;//排队中['group'=>[]]
    public $alias=[];//['name'=> [pid,conn] ...]//别名进程
    public $clients =[];//['addr'=>conn]
    public $ping = [];
    public function onWorkerStart()
    {
        $event = \Workerman\Worker::$globalEvent;
        $event->add(1, \Workerman\Events\EventInterface::EV_TIMER, function () {
            // // //内存监控
            if (isset($this->option['profile'])) {
                $alias=[];
                foreach ($this->alias as $key => $value) {
                    $alias[$key] =[$value[0] , $value[1]->getRemoteIp().':'.$value[1]->getRemotePort() ];
                }
                file_put_contents($this->option['profile'], json_encode([
                    'time'=>date('Y-m-d H:i:s'),
                    'opt'=>$this->option,
                    '------------------------'=>'',
                    'count'=>$this->worker_count,
                    'idle'=>count($this->idle_workers),
                    'task'=>count($this->queue_tasks),
                    'alias'=> $alias ,
                    'ping'=>$this->ping,
                ], JSON_PRETTY_PRINT));
            }

            //进程池调度
            switch (true) {
                case count($this->queue_tasks)>0 && $this->worker_count < $this->option['proc_max']:
                    Util::forkMethod(DispatchClient::class.'::startWorker', $this->option);
                    break;
                case $this->worker_count < $this->option['proc_min']:
                    Util::forkMethod(DispatchClient::class.'::startWorker', $this->option);
                    break;
            }
        });
        //进程池调度
        for ($i=0; $i<$this->option['proc_min']; $i++) {
            Util::forkMethod(DispatchClient::class.'::startWorker', $this->option);
        }
        
        //echo "\n---------------\nBG STARTED\n---------------\n";
    }
    public function onConnect($conn)
    {
        $this->clients[$conn->getRemoteIp().':'.$conn->getRemotePort()] = $conn;
    }
    public function onClose($conn)
    {
        unset($this->clients[$conn->getRemoteIp().':'.$conn->getRemotePort()]);
        unset($this->ping[$conn->getRemoteIp().':'.$conn->getRemotePort()]);
        foreach ($this->alias as $key => $c) {
            if ($c[1]===$conn) {
                unset($this->alias[$key]);
            }
        }
        foreach ($this->idle_workers as $key => $c) {
            if ($c===$conn) {
                unset($this->idle_workers[$key]);
                $this->worker_count--;
                break;
            }
        }
    }
    public function onMessage($conn, $data)
    {
        //echo "\nS:".substr(json_encode($data),0,128) ;
        switch ($data[0]) {
            case 'worker':
                $this->worker_count++;
                $this->idle_workers->push($conn);
                $this->dispatchTask();
                break;
            case 'done'://0:TOKEN
                $this->idle_workers->push($conn);
                $this->dispatchTask();
                break;
            case 'ping':
                $this->ping[$conn->getRemoteIp().':'.$conn->getRemotePort()] = $data[1];
                break;
            case 'spawn'://0:TOKEN 1:FN 2:ARGS 3:OPT
                $this->queue_tasks->push( $data );
                $this->dispatchTask();
                break;
            case 'send'://0:TOKEN 1:NAME 2:DATA
                if (isset($this->alias[$data[1]])) {
                    list($pid,$cc) = $this->alias[$data[1]];
                    $cc->send(['send',$pid,$data[2]]);
                }
                break;
            case 'register'://0:TOKEN 1:UNAME 2:PID
                //注册名字远程唯一覆盖
                $this->alias[$data[1]]=[$data[2],$conn] ;
                break;
            case 'unregister'://0:TOKEN 1:UNAME 2:PID
                //不能注销其他链接注册的名字
                if (isset( $this->alias[$data[1]] )
                && $this->alias[$data[1]][0] == $data[2] ) {
                    unset($this->alias[$data[1]]);
                }
                break;
            default:
                throw new \Exception(print_r($data, true), 1);
        }
    }
 
    public function dispatchTask()
    {
        if (count($this->queue_tasks) && count($this->idle_workers)) {
            $d = $this->queue_tasks->pop();
            $c = $this->idle_workers->pop();
            $c->send($d);
        }
    }
}
